package top.suven.base.core.mq.rocketmq;

import org.slf4j.MDC;
import top.suven.base.core.redis.RedisClusterServer;
import top.suven.base.core.redis.RedisKeys;
import top.suven.base.http.enums.SysMsgEnumType;
import top.suven.base.http.exception.SystemRuntimeException;
import top.suven.base.util.constants.GlobalConstants;
import top.suven.base.util.utilize.GlobalIdGenerator;
import top.suven.base.util.utilize.JsonUtils;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.stereotype.Component;

/**
 * @ClassName:
 * @Description:
 * @Author lixiangling
 * @Date 2018/5/22 17:12
 * @Copyright: (c) 2018 gc by https://www.gc.com.cn/
 * @Version : 1.0.0
 * --------------------------------------------------------
 * modifyer    modifyTime                 comment
 * <p>
 * --------------------------------------------------------
 */
@Component
@ConditionalOnClass(RocketMQAutoConfig.class)
public class RocketMQProducerHandler<T> {
    private Logger logger = LoggerFactory.getLogger(RocketMQProducerHandler.class);


    @Autowired
    private RedisClusterServer redisClusterServer;
    @Autowired
    private RocketThreadPool rocketThreadPool;
    /**
     * @Title: 处理消息的接口
     * @Description:
     * @param
     * @return
     * @throw
     * @author lixiangling
     * @date 2018/5/22 17:18
     *  --------------------------------------------------------
     *  modifyer    modifyTime                 comment
     *
     *  --------------------------------------------------------
     */
    public void sendMessage(String topicName,T msg){
        DefaultMQProducer defaultMQProducer = RocketMQProducerFactory.getMQProducer(topicName);
        if(defaultMQProducer == null){
            throw new SystemRuntimeException(SysMsgEnumType.MQ_ROCKET_PRODUCER_NOT_FOUND);
        }
        RocketMqTopic topic = RocketMQProducerFactory.getTopName(topicName);
        if(topic == null){
            throw new SystemRuntimeException(SysMsgEnumType.MQ_ROCKET_TOPIC_NOT_FOUND);
        }
        long globalId = GlobalIdGenerator.get(topic.getModule()).nextId();
        String traceId =  MDC.get(GlobalConstants.LOGBACK_TRACE_ID);
        Message message = new Message();
        message.setTopic(topic.getTopicName());
        message.setTags(topic.getTag());
        message.setDelayTimeLevel(0);
        message.setKeys(topicName+globalId);
        message.putUserProperty(GlobalConstants.LOGBACK_TRACE_ID,traceId);
        //拼装消息队列
        RocketMQMessage<T> rocketMQMessage = new RocketMQMessage<T>();
        rocketMQMessage.setGlobalId(globalId);
        rocketMQMessage.setRetryCount(topic.getFailedRetryCount());
        rocketMQMessage.setBody(msg);
        //pb序列化对象
        //byte[] prefixByte = SerializableUtil.toBytes(rocketMQMessage);
        String jsonRocketMqMessage = JsonUtils.toJsonString(rocketMQMessage);
        message.setBody(jsonRocketMqMessage.getBytes());
        redisClusterServer.setex(RedisKeys.MQ_CONSUMER+rocketMQMessage.getGlobalId(), String.valueOf(topic.getFailedRetryCount()),RocketMQListener.CACHE_TIME);


        logger.info("send rocketMq nameSrv=【{}】 message= 【{}】", defaultMQProducer.getNamesrvAddr(),jsonRocketMqMessage);
        RocketMQThread rocketMQThread = new RocketMQThread(message, defaultMQProducer);
        rocketThreadPool.run(rocketMQThread);
    }

    /**
     * @Title: 处理消息的接口
     * @Description: 消息处理失败补偿重发专用
     * @param
     * @return
     * @throw
     * @author lixiangling
     * @date 2018/5/22 17:18
     *  --------------------------------------------------------
     *  modifyer    modifyTime                 comment
     *
     *  --------------------------------------------------------
     */
    public void sendCompensateMessage(String topicName,long globalId,T msg){
        DefaultMQProducer defaultMQProducer = RocketMQProducerFactory.getMQProducer(topicName);
        if(defaultMQProducer == null){
            throw new SystemRuntimeException(SysMsgEnumType.MQ_ROCKET_PRODUCER_NOT_FOUND);
        }
        RocketMqTopic topic = RocketMQProducerFactory.getTopName(topicName);
        if(topic == null){
            throw new SystemRuntimeException(SysMsgEnumType.MQ_ROCKET_TOPIC_NOT_FOUND);
        }
        Message message = new Message();
        message.setTopic(topic.getTopicName());
        message.setTags(topic.getTag());
        message.setDelayTimeLevel(0);
        message.setKeys("CMT_"+topicName+globalId);
        //拼装消息队列
        RocketMQMessage<T> rocketMQMessage = new RocketMQMessage<T>();
        rocketMQMessage.setGlobalId(globalId);
        rocketMQMessage.setRetryCount(topic.getFailedRetryCount());
        rocketMQMessage.setBody(msg);
        //pb序列化对象
        //byte[] prefixByte = SerializableUtil.toBytes(rocketMQMessage);
        String jsonRocketMqMessage = JsonUtils.toJsonString(rocketMQMessage);
        message.setBody(jsonRocketMqMessage.getBytes());
        redisClusterServer.setex(RedisKeys.MQ_CONSUMER+rocketMQMessage.getGlobalId(), String.valueOf(topic.getFailedRetryCount()),RocketMQListener.CACHE_TIME);


        logger.info("send rocketMq nameSrv=【{}】 message= 【{}】", defaultMQProducer.getNamesrvAddr(),jsonRocketMqMessage);
        RocketMQThread rocketMQThread = new RocketMQThread(message, defaultMQProducer);
        rocketThreadPool.run(rocketMQThread);
    }
}
